package com.dstation.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dstation.domain.MqMessage;
import com.dstation.service.interf.IRecommendService;
import com.dstation.utils.Constants;
import com.dstation.utils.MQConstant;
import com.dstation.utils.RecommendStrategyUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Service;
import org.apache.rocketmq.common.message.MessageExt;
import javax.annotation.Resource;

@Service
@Slf4j
@RocketMQMessageListener(topic = MQConstant.RECOMMEND_TOPIC, selectorExpression = MQConstant.RECOMMEND_USER_BROWSER_TAG, consumerGroup = MQConstant.RECOMMEND_USER_BROWSER_GROUP)
public class BrowserMqConsumer  implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    @Resource
    private IRecommendService recommendService;


    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
        // 如果非第一次启动，那么按照上次消费的位置继续消费
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 设置为集群消费(区别于广播消费)
        // MQ默认集群消费
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        //设置最大重试次数
        defaultMQPushConsumer.setMaxReconsumeTimes(1);
        // 设置并发数
        defaultMQPushConsumer.setConsumeThreadMin(1);
        defaultMQPushConsumer.setConsumeThreadMax(20);
        log.info("====实时推荐-用户浏览  consumer=====");
    }

    @Override
    public void onMessage(MessageExt messageExt) {
        long msgTime = messageExt.getBornTimestamp();

        if (msgTime > RecommendStrategyUtil.REC_LIMIT_TIME) {
            try {

                System.out.println(messageExt);
                MqMessage message = JSONObject.parseObject(messageExt.getBody(), MqMessage.class);

                log.info("UserBrowseLogConsumer 开始处理 【START】, 请求的数据为 :{}", JSON.toJSONString(message));
                // 参数校验
                if (message == null) {
                    log.error("UserBrowseLogConsumer，参数校验未通过...");
                    return;
                }


                int browserClick = Constants.BROWSER_CLICK;
//
                recommendService.simlarRecForUserRandom( message, browserClick);
//
                log.info("用户{}    {}物品  实时推荐分配数据成功", message.getUserId());
            } catch (Exception e) {
                log.info("浏览数据消息消费失败");
            }


        } else {
            log.info("消息不在限制时间内，拒绝消费！");
        }
    }


}
